package com.atguigu.hbase;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * Created by Smexy on 2023/2/20
 *
 *  准则： 所有的客户端只要是连接hbase，只需要hbase存储元数据的zk集群地址即可。
 *
 *  ①创建一个客户端对象，连接服务端
 *          创建方式: 通过ConnectionFactory
 *          创建的连接对象Connection，代表一个独立的连接，连接zk，server，维护元数据的缓存，定位region
 *                  以上资源，可以被从 当前Connection中获取的Table和Admin对象共享。
 *
 *          由于Connection对象的创建是重量级(耗时，耗资源)，因此一个客户端应该只创建一次连接，在不同的线程中共享(线程安全)。
 *
 *
 *                  Table： 代表一张表。可以执行DML。
 *                  Admin: 代表管理员。可以执行DDL。
 *
 *            在创建Table和Admin的时候，由于它们的创建是轻量级(不费时，不费资源)的，且不是线程安全的。因此每个线程都拥有自己的Table和Admin的实例。
 *
 *  ②使用客户端，向服务端发送命令
 *      增，改:
 *          Table.put(Put t)
 *      查单行:
 *          Table.get(Get t)
 *       查多行:
 *          Table.getScanner(Scan t)
 *        删除:
 *          Table.delete(Delete d)
 *
 *       hbase没有数据类型，所有的数据都必须以byte[]进行存储。
 *       要写的数据，必须从 xxx 转换为 byte[].
 *
 *       工具类:
 *              Bytes:   帮你完成从 xxx 到 byte[]的相互转换。
 *                      Bytes.toBytes(Xxx ): 从xxx到byte[]
 *                      Bytes.toXxx(byte []): 从byte[]转换为xxx类型
 *              CellUtil: 帮你获取一个value的各种属性(rowkey,cf,cq,ts)
 *                      CellUtil.cloneXxx(): 获取xxx属性。
 *
 *  ③关闭客户端
 */
public class HBaseUtil
{
    //通过单例模式实现只创建一次
    private static Connection conn;

    static {
        try {
            //读取hadoop的配置文件和hbase的配置文件
            conn = ConnectionFactory.createConnection();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //用来遍历一行的所有列值
    public static void parseResult(Result result){

        //获取列值的集合
        Cell[] cells = result.rawCells();

        for (Cell cell : cells) {

            System.out.println(
                "rk:"+Bytes.toString(CellUtil.cloneRow(cell)) + " , " +
                 Bytes.toString(CellUtil.cloneFamily(cell)) + ":" +
                 Bytes.toString(CellUtil.cloneQualifier(cell)) + "=" +
                 Bytes.toString(CellUtil.cloneValue(cell)) + ",ts = " + cell.getTimestamp()
            );

        }

    }

    //构造Put对象  Put对象代表一次写入操作   put 表名,rowkey,cf:cq,value
    public static Put getPut(String rk,String cf,String cq,String value){

        //一个put对象向一行执行写入
        Put put = new Put(Bytes.toBytes(rk));
        //需要向哪些列写入，通过put.addColumn()操作
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes(cq),Bytes.toBytes(value));
        return put;
    }

    //根据表名创建一个表对象
    public static Table getTable(String name) throws IOException {

        if (StringUtils.isBlank(name)){
            throw new RuntimeException("表名非法!");
        }
        Table table = conn.getTable(TableName.valueOf(name));
        return table;

    }

    public static void close() throws IOException {
        if (conn != null){
            conn.close();
        }
    }



}
